package com.google.code.p.restrpc.client.loadbalance;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.google.code.p.restrpc.RpcRequest;
import com.google.code.p.restrpc.client.loadbalance.CachedWeightedRoundRobinLoadBalanceStrategy.AtomicCounter;
import com.google.code.p.restrpc.core.url.DefaultRestfulRequest;

public class WeightedRoundRobinLoadBalanceStrategy extends
                                                  RoundRobinLoadBalanceStrategy {
    
    private static final Log logger = LogFactory.getLog(WeightedRoundRobinLoadBalanceStrategy.class);
    
    // private final AtomicInteger lastSelected = new
    // AtomicInteger(Integer.MAX_VALUE);
    //    
    // private final AtomicInteger lastWeight = new
    // AtomicInteger(Integer.MAX_VALUE);
    
    private final AtomicCounter atomicCounter = new AtomicCounter();
    
    private ReentrantLock lock = new ReentrantLock();
    
    @Override
    protected int doSelect(String serviceName,
                           String methodName,
                           RpcRequest[] requests) {
        
        int maxWeight = requests[0].getWeight();
        int minWeight = requests[0].getWeight();
        for (int i = 1; i < requests.length; i++) {
            maxWeight = Math.max(maxWeight, requests[i].getWeight());
            minWeight = Math.min(minWeight, requests[i].getWeight());
        }
        
        if ((minWeight == maxWeight) || (maxWeight == 0) || (minWeight == 0)) {
            if (logger.isWarnEnabled()) {
                logger.warn("WeightedRoundRobinLoadBalanceStrategy's prerequisites is not satified and RoundRobinLoadBalanceStrategy will work instead!");
            }
            return super.doSelect(serviceName, methodName, requests);
        }
        
        atomicCounter.weight.compareAndSet(Integer.MAX_VALUE, 0);
        
        int calculatedWeight = 0;
        int selectedIndex = 0;
        List<Integer> matched = new ArrayList<Integer>(requests.length);
        try {
            lock.lock();
            
            int weightValue = atomicCounter.weight.get();
            if (weightValue > 0) {
                calculatedWeight = weightValue % maxWeight;
            }
            
            for (int i = 0; i < requests.length; i++) {
                if (requests[i].getWeight() > calculatedWeight) {
                    matched.add(i);
                }
            }
            
            // initialize a random value to avoid instantaneous peak to server
            // when it startup
            atomicCounter.selected.compareAndSet(Integer.MAX_VALUE,
                                                 new Random().nextInt(matched.size()));
            
            selectedIndex = atomicCounter.selected.get() % matched.size();
            if (selectedIndex >= (matched.size() - 1)) {
                // if the index equals to matched size,it means current loop
                // ends,next loop will start.
                // the selected will reset to zero
                // and the weight counter must increment
                atomicCounter.weight.compareAndSet(weightValue, weightValue + 1);
                atomicCounter.selected.set(0);
            }
            else {
                atomicCounter.selected.incrementAndGet();
            }
        }
        finally {
            lock.unlock();
        }
        return matched.get(selectedIndex).intValue();
    }
    
    static class AtomicCounter {
        
        /**
         * always increment it
         */
        final AtomicInteger selected = new AtomicInteger(Integer.MAX_VALUE);
        
        /**
         * the weight used to filter the item which >= the weight, when all item
         * iterated,increment the weight to next loop.
         */
        final AtomicInteger weight = new AtomicInteger(Integer.MAX_VALUE);
        
    }
    
    public static void main(String[] args) {
        DefaultRestfulRequest[] requests = new DefaultRestfulRequest[5];
        requests[0] = new DefaultRestfulRequest();
        requests[0].setWeight(5);
        requests[1] = new DefaultRestfulRequest();
        requests[1].setWeight(4);
        requests[2] = new DefaultRestfulRequest();
        requests[2].setWeight(3);
        requests[3] = new DefaultRestfulRequest();
        requests[3].setWeight(2);
        requests[4] = new DefaultRestfulRequest();
        requests[4].setWeight(2);
        
        LoadBalanceStrategy loadBalance = new WeightedRoundRobinLoadBalanceStrategy();
        for (int i = 0; i < 100; i++) {
            System.out.println(i + ":"
                               + loadBalance.select("testService",
                                                    "testMethod",
                                                    requests));
        }
    }
}
